package com.chinamobile.iot.mqtt.spi.impl;

import com.alibaba.fastjson.JSON;
import com.chinamobile.iot.command.CommandCache;
import com.chinamobile.iot.config.KafkaTopicConfiguration;
import com.chinamobile.iot.config.MqttConfiguration;
import com.chinamobile.iot.kafka.KafkaDataEventProducer;
import com.chinamobile.iot.kafka.KafkaDataProducer;
import com.chinamobile.iot.kafka.KafkaConnEventProducer;
import com.chinamobile.iot.mqtt.interception.MessageHandler;
import com.chinamobile.iot.mqtt.spi.model.StoredMessage;
import com.chinamobile.iot.mqtt.spi.processor.ClientEvent;
import com.chinamobile.iot.mqtt.spi.processor.DataProcessor;
import io.netty.handler.codec.mqtt.MqttQoS;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * Created by szl on 2017/7/31.
 */
@Component("dataHandler")
public class DataHandler implements MessageHandler {

    private static Logger logger = LoggerFactory.getLogger(DataHandler.class);
    @Autowired
    private KafkaTopicConfiguration kafkaTopic;
    @Autowired
    private CommandCache commandCache;
    @Autowired
    private KafkaDataProducer kafkaDataProducer;
    @Autowired
    private KafkaConnEventProducer kafkaConnEventProducer;
    @Autowired
    private KafkaDataEventProducer kafkaDataEventProducer;
    @Autowired
    private MqttConfiguration mqttConfig;
    @Autowired
    private DataProcessor dataProcessor;
    @Autowired
    private ProtocolProcessor protocolProcessor;

    @Override
    public boolean init() {
        return true;
    }

    @Override
    public void destroy() {

    }

    @Override
    public boolean onConnecting(String clientId, String mqttId) {
        return true;
    }

    @Override
    public void onConnected(String clientId, String mqttId, boolean cleanSession) {
        if (dataProcessor.needToSaveEvent(clientId, mqttId) && mqttConfig.isKafka()) {
            Object event = dataProcessor.createConnEvent(clientId, mqttId, ClientEvent.CONNECT);
            if (event != null) {
                kafkaConnEventProducer.sendEvent(mqttId, kafkaTopic.getConnEvent(), event);
                logger.info("send event to kafka : {}", JSON.toJSON(event));
            }
        }
    }

    @Override
    public void onClientDisconnected(String clientId, String mqttId, boolean cleanSession) {
        sendDisconnectToKafka(clientId, mqttId);
    }

    @Override
    public void onTopicPublished(String clientId, String mqttId, StoredMessage msg, String filter, int subCount) {
        if (clientId == null){
            return;
        }
        String topic = msg.getTopic();
        byte[] bytes = msg.getPayload();
        MqttQoS qos = msg.getQos();

        // 如果数据需要保存就发送到kafka
        if (dataProcessor.needToSaveData(clientId, mqttId, topic, bytes)) {
            Object data = dataProcessor.createData(clientId, mqttId, topic, bytes);
            String dataId = dataProcessor.getDataId(clientId, mqttId, topic, bytes);
            if (data != null) {
                if (qos.value() == MqttQoS.AT_LEAST_ONCE.value()) {
                    kafkaDataProducer.sendData(dataId, kafkaTopic.data(), data, msg.getMessageId());
                } else {
                    kafkaDataProducer.sendData(dataId, kafkaTopic.data(), data);
                }
            }
        } else {
            if (qos.value() == MqttQoS.AT_LEAST_ONCE.value()) {
                protocolProcessor.sendPubAck(clientId, msg.getMessageId());
            }
        }

        if (dataProcessor.needToSaveEvent(clientId, mqttId)){
            Object event = dataProcessor.createDataEvent(clientId, mqttId, filter, 1);
            if (event != null){
                kafkaDataEventProducer.sendEvent(mqttId, filter, event);
            }
        }

    }

    @Override
    public void onTopicSubscribed(String clientId, String mqttId, String topicFilter, int qos, boolean cleanSession, int oldQos) {

    }

    @Override
    public void onTopicUnsubscribed(String topic, int qos, String clientId, boolean cleanSession) {

    }

    @Override
    public void onMessageAcknowledged(String clientId, String mqttId, int messageId) {
        String commandId = commandCache.getCommandId(clientId, mqttId, messageId);
        if (commandId != null) {
            // 加入响应，3秒后向Kafka反馈命令执行成功
            commandCache.addAck(clientId, mqttId, messageId, commandId);
        }
    }

    @Override
    public void onRouteMessageToClient(String clientId, String mqttId, String topic) {
        if (dataProcessor.needToSaveEvent(clientId, mqttId) && mqttConfig.isKafka()) {
            Object event = dataProcessor.createDataEvent(clientId, mqttId, topic, -1);
            if (event != null) {
                kafkaConnEventProducer.sendEvent(mqttId, kafkaTopic.getDataEvent(), event);
                logger.info("send rout sub message event to kafka : {}", JSON.toJSON(event));
            }
        }
    }

    private void sendDisconnectToKafka(String clientId, String mqttId) {
        if (dataProcessor.needToSaveEvent(clientId, mqttId) && mqttConfig.isKafka()) {
            Object event = dataProcessor.createConnEvent(clientId, mqttId, ClientEvent.DISCONNECT);
            if (event != null) {
                kafkaDataEventProducer.sendEvent(mqttId, kafkaTopic.getConnEvent(), event);
                logger.info("send event to kafka :  {}", JSON.toJSONString(event));
            }
        }
    }
}
